package com.hugh.commom.zk.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.hugh.commom.selector.ChannelFutureEntity;
import com.hugh.commom.selector.ChannelSelect;
import com.hugh.commom.zk.client.entity.ZkNodeEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @Author: Hugh
 * @Date: 2020/6/23
 */
@Component
@Slf4j
public class ZkRpcHandle {

    private static Charset charset = Charset.forName("utf-8");

    @Autowired
    private CuratorFramework client;

    @Autowired
    private ChannelSelect channelSelect;

    private List<String> services;

    public String regist(Map<String, ZkNodeEntity> pathMap) throws Exception {

        // 创建节点
        for(Map.Entry<String, ZkNodeEntity> entry : pathMap.entrySet()){
            List<ZkNodeEntity> znodeList = new ArrayList<>();
            String path = "/" + entry.getKey() + "/" + "providers";
            // 状态信息
            Stat stat = client.checkExists().forPath(path);
            if(Objects.nonNull(stat)) {
                byte[] bytes = client.getData().forPath(path);
                znodeList = JSONArray.parseArray(new String(bytes), ZkNodeEntity.class);
                znodeList.add(entry.getValue());
                client.setData().forPath(path, JSON.toJSONString(znodeList).getBytes());
            }else{
                znodeList.add(entry.getValue());
                client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL).forPath(path, JSON.toJSONString(znodeList).getBytes());
            }
        }
        return "";
    }


    /**
     * /org.apache.dubbo.demo.DemoService/providers/ip
     * key是org.apache.dubbo.demo.DemoService
     * val是ips
     * @return
     */
    public Map<String, List<ZkNodeEntity>> getProviderAddr(List<String> services) throws Exception {
        Map<String, List<ZkNodeEntity>> res = new HashMap<>();
        if(!CollectionUtils.isEmpty(services)){
            for(String key : services){
                List<ZkNodeEntity> znodeList = new ArrayList<>();
                StringBuilder sb = new StringBuilder();
                buildPath(key, sb);
                String path = sb.toString();
                // 状态信息
                Stat stat = client.checkExists().forPath(path);
                if(Objects.nonNull(stat)) {
                    byte[] bytes = client.getData().forPath(path);
                    znodeList = JSONArray.parseArray(new String(bytes), ZkNodeEntity.class);
                    res.put(key, znodeList);
                }else {
                    client.create().creatingParentsIfNeeded()
                            .withMode(CreateMode.EPHEMERAL).forPath(path, JSON.toJSONString(znodeList).getBytes());
                }
                //添加监听事件
                ZKWatch watch = new ZKWatch(sb.toString());
                byte[] buffer = client.getData().usingWatcher(watch).forPath(path);
                log.info(new String(buffer, charset));
            }
        }
        this.services = services;
        log.info("启动时加载zk节点数据:{}", JSON.toJSON(res));
        return res;
    }

    private static void buildPath(String key, StringBuilder sb){
        sb.append("/").append(key).append("/").append("providers");
    }

    private static String getPathKey(String path){
        String[] split = path.split("/");
        for(String string : split){
            if(string.startsWith("com")){
                return string;
            }
        }
        return "";
    }
    /**
     * 添加监控
     */
    public class ZKWatch implements CuratorWatcher {

        private final String path;

        public ZKWatch(String path) {
            this.path = path;
        }

        @Override
        public void process(WatchedEvent event) throws Exception {
            List<ZkNodeEntity> znodeList = new ArrayList<>();
            // 状态信息
            Stat stat = client.checkExists().forPath(path);
            if(Objects.isNull(stat)) {
                client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL).forPath(path, JSON.toJSONString(znodeList).getBytes());
            }
            //继续订阅该节点
            byte[] data = client.
                    getData().
                    usingWatcher(this).forPath(path);
            log.info("监听节点数据变更" + path + " : " + new String(data, Charset.forName("utf-8")));
            //获取节点数据
            byte[] bytes = client.getData().forPath(path);
            znodeList = JSONArray.parseArray(new String(bytes), ZkNodeEntity.class);

            channelSelect.updateChannelSelect(getPathKey(path), znodeList);
            log.info("***重要***节点更新，更新类型是:{},更新信息是:{}", event.getType(), new String(data, Charset.forName("utf-8")));
        }
    }

    /**
     * 更新响应时间
     *
     * @param channelFutureEntity
     * @throws Exception
     */
    public void upDateZkNodeRt(ChannelFutureEntity channelFutureEntity) throws Exception {
        List<ZkNodeEntity> znodeList = new ArrayList<>();
        StringBuilder sb = new StringBuilder();
        buildPath(channelFutureEntity.getService(), sb);
        String path = sb.toString();
        // 状态信息
        Stat stat = client.checkExists().forPath(path);
        if(Objects.nonNull(stat)) {
            byte[] bytes = client.getData().forPath(path);
            znodeList = JSONArray.parseArray(new String(bytes), ZkNodeEntity.class);
            if(!CollectionUtils.isEmpty(znodeList)){
                for(ZkNodeEntity zNode : znodeList){
                    if(zNode.getIp().equals(channelFutureEntity.getIp()) && zNode.getPort().equals(channelFutureEntity.getPort())){
                        zNode.setRt(channelFutureEntity.getRt());
                    }
                }
            }
            client.setData().forPath(path, JSON.toJSONString(znodeList).getBytes());
        }else {
            client.create().creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL).forPath(path, JSON.toJSONString(znodeList).getBytes());
        }
    }
}
